package com.pivotal.hawq.mapreduce.metadata;

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */


import com.pivotal.hawq.mapreduce.util.HAWQJdbcUtils;

import java.sql.Connection;
import java.sql.SQLException;

/**
 * Convenient class for extracting table's metadata from HAWQ. This class is
 * used internally by HAWQInputFormat and usually don't need to be used by user.
 * It is available in public API for users who want to use HAWQAOInputFormat and
 * HAWQParquetInputFormat directly.
 * <p>
 *
 * Metadata can be extracted in two ways:
 * <p>
 * <ul>
 *     <li>{@code MetadataAccessor.newInstanceUsingJDBC(...)} Connect to a running
 *     	   HAWQ and get metadata from system catalog.</li>
 *     <li>{@code MetadataAccessor.newInstanceUsingFile(...)} You must dump the metadata
 *         into a file using <i>hawq extract</i> in advance, then you can load the metadata
 *         from file.</li>
 * </ul>
 *
 * In either way, you can then call {@code getAOMetadata} or {@code getParquetMetadata}
 * with respect to the actual table format indicated by {@code getTableFormat}. Only
 * AO and Parquet tables are supported for now.
 */
public abstract class MetadataAccessor {

	/**
	 * Get the storage format of the input table.
	 * @return the storage format of the input table
	 */
	public abstract HAWQTableFormat getTableFormat();

	/**
	 * Get metadata for an AO table. If the input table is not AO,
	 * an IllegalStateException is threw.
	 * @return metadata of the AO table
	 */
	public abstract HAWQAOTableMetadata getAOMetadata();

	/**
	 * Get metadata for a Parquet table. If the input table is not Parquet,
	 * an IllegalStateException is threw.
	 * @return metadata of the Parquet table
	 */
	public abstract HAWQParquetTableMetadata getParquetMetadata();

	/**
	 * Connect to database and extract table metadata using JDBC. Only AO and Parquet table are supported.
	 * @param dbUrl		database connection string in form 'host:port/dbname'
	 * @param username	name of the database user
	 * @param password	password of the database user
	 * @param tableName	name of an AO or Parquet table, valid format are 'namespace_name.table_name' or
	 *                  simply 'table_name' if use the default 'public' namespace
	 * @return MetadataAccessor instance with input table's metadata
	 * @throws MetadataAccessException if failed to extract metadata
	 */
	public static MetadataAccessor newInstanceUsingJDBC(
			String dbUrl,
			String username,
			String password,
			String tableName) {

		Connection conn = null;
		try {
			conn = HAWQJdbcUtils.getConnection("jdbc:postgresql://" + dbUrl, username, password);
			return new MetadataSQLAccessor(conn, tableName);

		} catch (Exception e) {
			throw new MetadataAccessException("failed to extract metadata!", e);
		} finally {
			try {
				HAWQJdbcUtils.closeConnection(conn);
			} catch (SQLException e) {
				throw new MetadataAccessException("failed to close Connection!", e);
			}
		}
	}

	/**
	 * Reading table's metadata file generated by hawq extract utility.
	 * @param file Path to the metadata file
	 * @return MetadataAccessor instance with input table's metadata
	 * @throws MetadataAccessException  if failed to extract metadata
	 */
	public static MetadataAccessor newInstanceUsingFile(String file) {

		try {
			return new MetadataYAMLAccessor(file);

		} catch (IllegalStateException e) {
		  throw new IllegalStateException(e.getMessage());
		} catch (Exception e) {
			throw new MetadataAccessException("error occurred when reading " + file, e);
		}
	}
}
